package data.platform.store;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import data.platform.common.domain.AggregateItem;
import data.platform.common.service.command.KeyValueRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.data.redis.core.ReactiveValueOperations;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;


@ConditionalOnBean(name = "redisConfig")
@Slf4j
@Repository
@RequiredArgsConstructor
public class RedisRepositoryImpl implements KeyValueRepository<String, byte[]> {
    final ReactiveStringRedisTemplate redisTemplate;
    private ReactiveValueOperations<String, String> operations;

    private ObjectMapper objectMapper;

    @PostConstruct
    public void init() {
        objectMapper = new ObjectMapper();
        objectMapper.registerModule(new JavaTimeModule());
        operations = redisTemplate.opsForValue();
    }

    @Override
    public Mono<byte[]> get(String key) {
        return operations.get(key).flatMap(s -> Mono.just(s.getBytes(StandardCharsets.UTF_8)));
    }

    //   把对象转换为json字符串进行存储
    @Override
    public void put(String key, byte[] value) {
        try {
            String valueJson = new String(value, StandardCharsets.UTF_8);
            operations.set(key, valueJson).subscribe();
        } catch (Exception e) {
            log.error("Error saving entry in Redis, cause: {}, message: {}", e.getCause(), e.getMessage());
        }
    }

    @Override
    public void delete(String key) {
        operations.delete(key).subscribe();
    }

    @Override
    public Flux<Tuple2<String, byte[]>> getByPrefixKey(String prefixKey) {
        String key = prefixKey + "*";
        return redisTemplate.scan(ScanOptions.scanOptions().match(key).build())
                .flatMap(k -> get(k).map(bytes -> Tuples.of(k, bytes)));
    }
}